/*******************************************************************************
 * Copyright (c) 2010 Mikhail Garber.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser Public License (LGPL)
 * which accompanies this distribution
 * 
 * Contributors:
 *     Mikhail Garber - initial API and implementation
 ******************************************************************************/
package plumber.process;

import java.util.Calendar;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import plumber.common.HasSetup;
import plumber.common.Utils;
import plumber.common.WorkItem;
import plumber.component.ComponentManager;
import plumber.component.DefaultRetryStrategy;
import plumber.component.RetryStrategy;
import plumber.context.ContextConverter;
import plumber.context.DefaultContextConverter;
import plumber.context.FlowContext;
import plumber.coordinator.CoordinatorServer;
import plumber.flow.FlowManager;
import plumber.model.FlowStep;
import plumber.model.StepOutcome;

/**
 * core class in the workflow system. 
 * runs poller thread to select list of work items to work with;
 * defines and runs thread pool of worker threads to work with work items;
 * upon execution of each step, evaluates the result of execution and determines next Step to run
 */
public class WorkManager implements HasSetup {

	

	
	private CoordinatorServer coordinator;
	
	
	private FlowManager flowManager;
	
	
	private ComponentManager componentManager;
	
	/**
	 * retry strategy in case of Step failure
	 */
	public void setRetryStrategy(RetryStrategy retryStrategy) {
		this.retryStrategy = retryStrategy;
	}

	private RetryStrategy retryStrategy = new DefaultRetryStrategy();
	
	
	private FlowTerminator terminator;
	
	
	private ContextConverter contextConverter = new DefaultContextConverter();
	
	
	private String namespace;
	
	/**
	 * poller configuration
	 */
	private long pollerDelay = 1000L;
	
	private int maxPollSize = 20;
	

	public void setPollerDelay(long pollerDelay) {
		this.pollerDelay = pollerDelay;
	}

	public void setMaxPollSize(int maxPollSize) {
		this.maxPollSize = maxPollSize;
	}

	/**
	 * namespace used by this workflow installation
	 */
	public void setNamespace(String namespace) {
		this.namespace = namespace;
	}

	/**
	 * how to convert item level task context from/to storable text to/from Java objects
	 */
	public void setContextConverter(ContextConverter contextConverter) {
		this.contextConverter = contextConverter;
	}

	/**
	 * Terminator used to kill hanged workers
	 */
	public void setTerminator(FlowTerminator terminator) {
		this.terminator = terminator;
	}
	
	/**
	 * where to get work from
	 */
	public void setCoordinator(CoordinatorServer coordinator) {
		this.coordinator = coordinator;
	}

	/**
	 * where/how to execute individual steps
	 */
	public void setComponentManager(ComponentManager componentManager) {
		this.componentManager = componentManager;
	}

	/** 
	 * where to find work flows
	 */
	public void setFlowManager(FlowManager flowManager) {
		this.flowManager = flowManager;
	}

	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
	private SynchronousQueue<Runnable> queue = new SynchronousQueue<Runnable>();
	
	/**
	 * thread pool of workers
	 */
	private ThreadPoolExecutor executor = null;

	@Override
	public void setup() {
		if(StringUtils.isEmpty(namespace)) {
			throw new IllegalStateException("namespace is not defined");
		}
		setupThreadPool();
		setupPoller();
		logger.info("setup");
	}

	private void setupThreadPool() {
		executor = new ThreadPoolExecutor(20, 40, 2, TimeUnit.SECONDS, queue);

	}

	Thread poller;
	
	/**
	 * Poller thread runs periodically and asks Coordinator Server for work items
	 */
	private void setupPoller() {
		poller = new Thread() {
			@Override
			public void run() {
				while (true)
					try {
						Thread.sleep(pollerDelay);
						runnerBody();

					} catch(InterruptedException ie) {
						logger.warn("poller interrupted");
						return;
					} catch (Exception ex) {
						logger.warn("problem in runner", ex);
					}
			}
		};
		poller.setDaemon(true);
		poller.setName("runner");
		poller.start();
		logger.info("poller started");
	}

	/**
	 * body of poller thread
	 * @throws InterruptedException
	 */
	protected void runnerBody() throws InterruptedException {
		logger.info("running poller");
		List<String> candidates = coordinator.getCandidates(namespace, maxPollSize);
		logger.info("got candidates:" + candidates);
		for (WorkItem candidate : this.contextConverter.itemsFromStorableString(candidates)) {
			if (coordinator.claimCandidate(candidate.getId())) {
				FlowTask task = asRunnable(candidate);
				task.setStarted(new Date());
				terminator.register(task);
				executor.execute(task);
			}
		}
	}

	/**
	 * setup a Work Item as a runnable thread. 
	 * Asks FlowManager for step definitions,
	 * sets up runnable to execute this step
	 * @param item to execute
	 * @return executable task
	 */
	private FlowTask asRunnable(final WorkItem item) {
		
		String stepId = item.getFlowStepId();
		logger.info("finding step for step id:" + stepId);
		final FlowStep step = (stepId == null) ?
				flowManager.findFirstStepByFlowName(item.getFlowName()) :
				flowManager.findStepById(item.getFlowName(), stepId);
		Utils.validatePresent(step, "step for item:" + item);
		
		
		
		Runnable r = new Runnable() {
			@Override
			public void run() {
				FlowContext fc = null;
				try {										
					
					logger.info("running step:" + step + " with arg:" + item.getContext());
					fc = contextConverter.contextFromStorableString(item.getContext());
					Object result = componentManager.execute(
							step.getComponent(), step.getMethod(),
							fc);
					processSuccess(item.getId(), step, fc, result);
				} catch (Exception ex) {
					logger.warn("in runnable", ex);
					processException(item.getId(), step, fc, getCause(ex));
				}
			}
			
		};

		FlowTask ft = new FlowTask(r, null);
		ft.setItem(item);
		ft.setMaxSeconds(step.getMaxTimeSeconds());
		return ft;
	}

	/**
	 * recursively extract the cause of the exception
	 * @param ex
	 * @return the cause
	 */
	protected Exception getCause(Exception ex) {
		if(ex.getCause() == null) {
			return ex;
		} else {
			return getCause((Exception) ex.getCause());
		}
	}

	/**
	 * in case of the exception thrown during the step execution, determine next step and update the Coordinator accordingly
	 * @param itemId of the item we are dealing with
	 * @param step being currently executed
	 * @param context passed with this step
	 * @param exeption thrown during step execution
	 */
	private void processException(String itemId, FlowStep step, FlowContext context, Exception ex) {
		logger.info("procesing exception:" + context);
		String nextStepId = determineNextStep(step, context, ex);		
		int retryCount = context.getRetryCount();
		context.setRetryCount(++retryCount);
		
		// if there is no next step - we failed, otherwise we will retry
		WorkItem.Status statusOfOldItem = (nextStepId == null) ? WorkItem.Status.FAILED
				: WorkItem.Status.TO_RETRY;
		
		// if it is the same step, we are re-trying so find out next delay,
		// otherwise - as soon as possible
		Date executionDate = (step.getId().equals(nextStepId)) ? determineDelay(context) : new Date(); 
		
		coordinator.advanceWork(
				contextConverter.toStorableString(context), 
				itemId, nextStepId,
				statusOfOldItem,
				executionDate);
	}

	
	/**
	 * Determine delay for the retry of the Step in case of failure.
	 * RetryStrategy component allows for pluggable delay configuration.
	 * @param context
	 * @return Date when next execution is scheduled
	 */
	private Date determineDelay(FlowContext context) {
		int count = context.getRetryCount();
		logger.info("delaying for retry count:" + count);
		int delay = retryStrategy.getDelaySecondsForRetryCount(count);
		Calendar cal = Calendar.getInstance();
		cal.add(Calendar.SECOND, delay);
		logger.info("delayed until:" + cal.getTime());
		return cal.getTime();
	}

	/**
	 * Deal with results of successful execution of a Step: determine next Step to run and update the
	 * Coordinator accordingly
	 * @param itemId of an item we are working with
	 * @param step we just executed
	 * @param context of this item
	 * @param result of the Step execution
	 */
	private void processSuccess(String itemId, FlowStep step, FlowContext context, Object result) {
		logger.info("processing success:" + context);
		String nextStepId = determineNextStep(step, context, result);
		WorkItem.Status status = WorkItem.Status.DONE; // done with this step
		if (nextStepId == null) { // if there is no next step
			if (step.getOutcomes().size() == 0) { // and no outcomes defined, workflow is done
				status = WorkItem.Status.FINAL;
			} else {
				throw new RuntimeException(
						"outcomes are configured but no next step");
			}
		}
		coordinator.advanceWork(
				contextConverter.toStorableString(context)
				, itemId, nextStepId,
				status, new Date());
	}

	/**
	 * figure out next Step in case of successful execution of current Step
	 * @param step
	 * @param context
	 * @param result
	 * @return step id of the next Step
	 */
	private String determineNextStep(FlowStep step, FlowContext context, Object result) {
		context.setRetryCount(1);
		return determineNextStep(step, result);
	}

	/**
	 * determine next Step in case of failure of the current Step. If retry count for this Step
	 * is not exausted, simply return the id of the current Step so it will be retried.
	 * @param step
	 * @param context
	 * @param result - Exception thrown during execution of this step
	 * @return
	 */
	private String determineNextStep(FlowStep step, FlowContext context,
			Exception result) {

		int retryCount = context.getRetryCount();
		if (retryCount < step.getMaxRetryCount()) {
			logger.info("retrying step, count:" + retryCount + ", max:" + step.getMaxRetryCount());
			return step.getId();
		} else {
			logger.info("no more retrys");
		}
		return determineNextStep(step, result);
	}

	/**
	 * determine the next Step to run base on current step and result of its execution
	 * @param step 
	 * @param result
	 * @return step id of the next step
	 */
	private String determineNextStep(FlowStep step, Object result) {
		logger.info("determine next step for:" + step + ", " + result);
		Set<StepOutcome> outcomes = step.getOutcomes();
		logger.info("all outcomes:" + outcomes);
		Set<StepOutcome> matchedOutcomes = new HashSet<StepOutcome>();
		for (StepOutcome outcome : outcomes) {

			if (outcome.match(result)) {
				matchedOutcomes.add(outcome);
			}
		}
		if (matchedOutcomes.size() > 1) {
			throw new IllegalStateException("more than one outcome matched:"
					+ matchedOutcomes);
		}
		if (matchedOutcomes.size() == 0) {
			logger.info("no outcomes matched");
			return null;
		}
		logger.info("outcomes:" + matchedOutcomes);
		return matchedOutcomes.iterator().next().getNextStep().getId();
	}

	@Override
	/**
	 * shutdown this component
	 */
	public void shutdown() {
		if (executor != null) {
			executor.shutdown();
		}
		this.poller.interrupt();
	}

}
